package com.breeze.components.socket.redis;

import com.alibaba.fastjson2.JSON;
import com.breeze.components.socket.WebSocket;
import com.breeze.components.socket.memory.MemWebSocketManager;
import com.breeze.components.socket.redis.action.Action;
import com.breeze.components.socket.redis.action.BroadCastAction;
import com.breeze.components.socket.redis.action.RemoveAction;
import com.breeze.components.socket.redis.action.SendMessageAction;
import com.breeze.components.socket.util.WebSocketUtil;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;

/**
 * WebSocket的session无法序列化,所以session还是保存在本地内存中，发送消息这种就走订阅发布模式
 *
 * @author breeze
 */
public class RedisWebSocketManager extends MemWebSocketManager {

  public static final String CHANNEL = "websocket";
  private static final String COUNT_KEY = "RedisWebSocketManagerCountKey";
  protected StringRedisTemplate stringRedisTemplate;

  public RedisWebSocketManager(StringRedisTemplate stringRedisTemplate) {
    this.stringRedisTemplate = stringRedisTemplate;
  }

  @Override
  public void put(String identifier, WebSocket webSocket) {
    super.put(identifier, webSocket);
    // 在线数量加1
    countChange(1);
  }

  @Override
  public void remove(String identifier) {
    boolean containsKey = localWebSocketMap().containsKey(identifier);
    if (containsKey) {
      super.remove(identifier);
    } else {
      Map<String, Object> map = new HashMap<>(2);
      map.put(Action.ACTION, RemoveAction.class.getName());
      map.put(Action.IDENTIFIER, identifier);
      // 在websocket频道上发布发送消息的消息
      stringRedisTemplate.convertAndSend(getChannel(), JSON.toJSONString(map));
    }
    // 在线数量减1
    countChange(-1);
  }

  @Override
  public void clearClientCount() {
    countChange(-localWebSocketMap().size());
  }

  @Override
  public int size() {
    return getCount();
  }

  @Override
  public void sendMessage(String identifier, String message) {
    WebSocket webSocket = get(identifier);
    // 本地能找到就直接发
    if (Objects.nonNull(webSocket)) {
      WebSocketUtil.sendMessage(webSocket.getSession(), message);
      return;
    }

    Map<String, Object> map = new HashMap<>(3);
    map.put(Action.ACTION, SendMessageAction.class.getName());
    map.put(Action.IDENTIFIER, identifier);
    map.put(Action.MESSAGE, message);
    // 在websocket频道上发布发送消息的消息
    stringRedisTemplate.convertAndSend(getChannel(), JSON.toJSONString(map));
  }

  @Override
  public void broadcast(String message) {
    Map<String, Object> map = new HashMap<>(2);
    map.put(Action.ACTION, BroadCastAction.class.getName());
    map.put(Action.MESSAGE, message);
    // 在websocket频道上发布广播的消息
    stringRedisTemplate.convertAndSend(getChannel(), JSON.toJSONString(map));
  }

  protected String getChannel() {
    return CHANNEL;
  }

  protected String getCountKey() {
    return COUNT_KEY;
  }

  /** 增减在线数量 */
  private void countChange(int delta) {
    ValueOperations<String, String> value = stringRedisTemplate.opsForValue();

    // 获取在线当前数量
    int count = getCount(value);

    count = count + delta;

    // 设置新的数量
    value.set(getCountKey(), String.valueOf(Math.max(count, 0)));
  }

  /** 获取当前在线数量 */
  private int getCount() {
    ValueOperations<String, String> value = stringRedisTemplate.opsForValue();
    return getCount(value);
  }

  private int getCount(ValueOperations<String, String> value) {
    String countStr = value.get(getCountKey());
    int count = 0;
    if (null != countStr) {
      count = Integer.parseInt(countStr);
    }
    return count;
  }
}
